package tableapi.udf

import bean.SensorReading
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.table.api.scala.{StreamTableEnvironment, _}
import org.apache.flink.table.api.{EnvironmentSettings, Table}
import org.apache.flink.table.functions.{AggregateFunction, TableAggregateFunction}
import org.apache.flink.types.Row
import org.apache.flink.util.Collector

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/11/28 23:10
  */
object Test4_TableAggregateFunction {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    //file source
    val inputPath: String = "src/main/resources/sensor.txt"
    val fileStream: DataStream[String] = env.readTextFile(inputPath)

    val socketStream = env.socketTextStream("192.168.88.106", 7777)

    //先转换为特定的类型
    val dataStream: DataStream[SensorReading] = fileStream.map(data => {
      val arr = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
        override def extractTimestamp(t: SensorReading): Long = t.timestamp * 1000L
      })

    val settings: EnvironmentSettings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()

    //创建表执行环境
    val tableEnv = StreamTableEnvironment.create(env, settings)

    val sensorTable: Table = tableEnv.fromDataStream(dataStream, 'id, 'temperature, 'timestamp.rowtime as 'ts)

    val top2Temp = new Top2Temp()
    val resultTabel: Table = sensorTable
      .groupBy('id)
      .flatAggregate(top2Temp('temperature) as ('temp,'rank))
      .select('id, 'temp,'rank)

    //sql


    resultTabel.toRetractStream[Row].print("result")

    env.execute("table aggregate test")


  }

  //定义一个类表示聚合状态
  class Top2TempAcc {
    var highestTemp: Double = Double.MinValue
    var secondHighestTemp: Double = Double.MinValue
  }

  //自定义表聚合函数，提取所有温度中温度值最高的两个温度
  class Top2Temp extends TableAggregateFunction[(Double, Int), Top2TempAcc] {
    override def createAccumulator(): Top2TempAcc = new Top2TempAcc()

    //实现计算聚合结果的函数accumulate
    def accumulate(acc: Top2TempAcc, temp: Double): Unit = {
      //判断当前温度值，是否比状态中的值大
      if (temp > acc.highestTemp) {
        //如果比最高温度还高，排在第一位，原来的第一位顺到第二位
        acc.secondHighestTemp = acc.highestTemp
        acc.highestTemp = temp
      } else if (temp > acc.secondHighestTemp) {
        //如果在最高和第二高之间，那么直接替换第二高位置
        acc.secondHighestTemp = temp
      }
    }

    //实现一个输出结果的方法，最终处理完表中所有数据时调用
    def emitValue(acc: Top2TempAcc, out: Collector[(Double, Int)]): Unit = {
      out.collect((acc.highestTemp, 1))
      out.collect((acc.secondHighestTemp, 2))
    }

  }


}
